package com.example.curator.async;

import com.example.curator.CuratorClientUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.x.async.AsyncCuratorFramework;
import org.apache.curator.x.async.AsyncEventException;
import org.apache.curator.x.async.WatchMode;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.junit.BeforeClass;
import org.junit.Test;

import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;


public class AsyncCuratorOperatorTest {

    private static CuratorFramework curatorFramework;

    @BeforeClass
    public static void getInstance( ) {
        curatorFramework = CuratorClientUtils.getInstance();
        System.out.println("连接成功.........");
    }

    /**
     * 异步操作
     */
    @Test
    public void async( ) throws Exception {
        AsyncCuratorFramework async = AsyncCuratorFramework.wrap(curatorFramework);
        String path = "/async";
        CompletionStage<String> stringCompletionStage = async.create().withMode(CreateMode.EPHEMERAL)
                .forPath(path, "12".getBytes())
                .whenComplete((name, exception) -> {
                    if (exception != null) {
                        exception.printStackTrace();
                    } else {
                        handleWatchedStage(async.watched().checkExists().forPath(path).event());
                        System.out.println(name);
                    }
                });
        String s = stringCompletionStage.toCompletableFuture().get();
        System.out.println(s);
    }

    @Test
    public void createThenWatchSimple( ) throws Exception {
        AsyncCuratorFramework async = AsyncCuratorFramework.wrap(curatorFramework);
        createThenWatchSimple(async, "/a");
    }

    private static void handleWatchedStage(CompletionStage<WatchedEvent> watchedStage) {
        // Watchers的异步处理很复杂，因为观察者可以多次触发并且 CompletionStage 不支持这个行为
        watchedStage.thenAccept(event -> {
            System.out.println(event.getType());
            System.out.println(event);
        });

        //如果存在连接问题，在观察者触发发信号通知连接问题的情况下，将异常地调用。
        // 必须调用“reset（）”来重置观看阶段
        watchedStage.exceptionally(exception -> {
            AsyncEventException asyncEx = (AsyncEventException) exception;
            asyncEx.printStackTrace();    // handle the error as needed
            handleWatchedStage(asyncEx.reset());
            return null;
        });
    }


    public static void createThenWatchSimple(AsyncCuratorFramework async, String path) throws ExecutionException, InterruptedException {
        CompletionStage<String> stringCompletionStage = async.create().forPath(path).whenComplete((name, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
            } else {
                async.with(WatchMode.successOnly).watched().checkExists().forPath(path).event().thenAccept(event -> {
                    System.out.println(event.getType());
                    System.out.println(event);
                });
            }
        });
        String s = stringCompletionStage.toCompletableFuture().get();
        System.out.println(s);
    }
}
